package com.example.order.messaging.consumer;

import com.example.order.messaging.message.OrderPaidMessage;
import com.example.order.service.OrderService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * 订单支付成功事件消费者
 * 监听 order_paid Topic
 * 
 * @author mall-team
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class OrderPaidConsumer {

    private final OrderService orderService;

    /**
     * 消费订单支付成功事件
     * 更新订单状态为已支付
     * 
     * @param message 支付成功消息
     * @param ack 手动确认对象
     */
    @KafkaListener(topics = "order_paid", groupId = "order-service-group")
    public void handleOrderPaid(OrderPaidMessage message, Acknowledgment ack) {
        log.info("收到订单支付成功事件: orderNo={}, paymentNo={}", 
            message.getOrderNo(), message.getPaymentNo());
        
        try {
            // 更新订单状态为已支付
            orderService.updateOrderStatusToPaid(message.getOrderNo(), message.getPaymentNo());
            
            log.info("订单状态更新成功: orderNo={}, status=PAID", message.getOrderNo());
            
            // 手动确认消息
            ack.acknowledge();
            
        } catch (Exception e) {
            log.error("处理订单支付事件失败: orderNo={}, error={}", 
                message.getOrderNo(), e.getMessage(), e);
            // 不确认消息，等待重试
            // 实际生产环境中可能需要更复杂的重试和死信队列机制
        }
    }
}

